package org.webscale.util;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.webscale.ApplicationException;

public class MultiThreadedClient<V> {
	private static final Logger log = Logger
			.getLogger(MultiThreadedClient.class);
	private int poolSize = 1;
	private ExecutorService pool;

	public MultiThreadedClient() {
		super();
	}

	private Future processJob(Callable job) {
		Future futureTask = null;
		if (!pool.isShutdown()) {
			log.debug("processing new client request ");
			futureTask = pool.submit(job);
			// pool.execute(client);
		} else {
			log.warn("The Pool is shutdown and hence will not run this task ");
		}
		return futureTask;

	}

	public void init() {
		pool = Executors.newFixedThreadPool(getMaxPoolSize()); // TODO: figure outut if you
																// want to use
																// cached pool
																// or fixed pool
		log.info("Init for MultiThreadedClient called");
	}

	public Job<V> createJob(){
		Job<V> job = new Job<V>();
		return job;
	}
	
	public int getPoolSize() {
		return poolSize;
	}

	public void setPoolSize(int poolSize) {
		this.poolSize = poolSize;
	}

	public int getMaxPoolSize() {
		return poolSize;
	}

	public void shutdown() throws InterruptedException {
		log.info("shoutdown called");
		pool.shutdown();
		pool.awaitTermination(30, TimeUnit.SECONDS);
		pool.shutdownNow();
	}

	public class Job<V> {
		ArrayList<Callable<V>> jobs = new ArrayList<Callable<V>>();
		private ArrayList<Future<V>> responses = new ArrayList<Future<V>>();

		public Job(){
			super();
		}
		
		public void addToJob(Callable<V> job) {
			jobs.add(job);
		}

		public void submitAllJobs() {
			for (Callable<V> job : jobs) {
				Future<V> f = processJob(job);
				responses.add(f);
			}
		}

		public List<V> getAllResponses() {
			ArrayList<V> responseList = new ArrayList<V>();
			Iterator<Future<V>> it = responses.iterator();
			while (it.hasNext()) {
				Future<V> f = it.next();
				V response;
				try {
					response = f.get();
					responseList.add(response);
				} catch (InterruptedException e) {
					log.error("Thread Interuppted - Exiting app", e);
					throw new ApplicationException(e);
				} catch (ExecutionException e) {
					log.error("Thread ExecutionException - Exiting app", e);
					throw new ApplicationException(e);
				}
			}
			return responseList;
		}

	}

}
